package com.example.mq.mqserver.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.VirtualHost;
import lombok.Data;

import java.util.Queue;
import java.util.concurrent.*;

/**
 * 实现消费者的回调消息
 */
@Data
public class ConsumerManager {
    private VirtualHost parent;
    //线程池
    private ExecutorService workPool = Executors.newFixedThreadPool(4);
    //阻塞队列 存放有消息的队列名
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    //扫描线程
    private Thread scannerThread = null;



    //创建一个扫描线程。该线程会持续从tokenQueue里获取队列名称（令牌）
    // 接着依据这个名称从内存数据中心获取对应的消息队列。
    // 若队列存在，就会对队列中的消息进行消费；若队列不存在，则会抛出异常。
    
    public ConsumerManager(VirtualHost p) {
        parent = p;
        scannerThread = new Thread(() -> {
            while (true) {
                try {
                    //拿到令牌
                    String queueName = tokenQueue.take();
                    //根据令牌找到消息队列
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    //判空
                    if (queue == null) {
                        throw new MqException("[ConsumerManager] 取令牌后，该消息队列不存在！ queueName=" + queueName);
                    }
                    //消费消息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();

                }
            }
        });
        // 把线程设为后台线程.
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    /**
     * 消费者消费消息
     * @param queue 从哪个队列中取出
     */
    public void consumeMessage(MSGQueue queue) {
        //按照轮询 取出一个消费者
        ConsumerEnv luckyDog = queue.chooseConsumerEnv();
        if (luckyDog == null) {
            //消费者为空 暂不需要消费 返回
            return;
        }
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            //消息队列为空 暂时不能消费 返回
            return;
        }
        //将消息带入到消费者的回调函数中，给线程池处理
        workPool.submit(() -> {
            try{
                //将消息放入待确认应答消息队列中 避免消息丢失
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                //真正 执行回调函数
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(), message.getBody());
                //消息是否为自动应答
                if (luckyDog.isAutoAck()) {
                    //消息是否可持久化
                    if(message.getDeliverMode() == 2){
                        //删除硬盘上的消息
                        parent.getDiskDataCenter().deleteMessage(queue,message);
                    }
                    //删除可持久化队列中的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(),message.getMessageId());
                    //删除内存中的消息
                    parent.getMemoryDataCenter().sendMessage(message,queue);
                    System.out.println("[ConsumerManager] 消息被成功消费! queueName=" + queue.getName());
                }

            }catch (Exception e){
                e.printStackTrace();
            }

            System.out.println("[COnsumerManager] 消息被成功消费 queueName= " + queue.getName());
        });

    }


    /**
     * 创建消费者
     * @param consumerTag   消费者标识
     * @param queueName 队列名称
     * @param autoAck 是否能自动删除
     * @param consumer 消费者
     */
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        //调用父类VirtualHost 的方法
        //查询时向哪个队列加入
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        //判空  为空队列不存在
        if (queue == null) {
            System.out.println("[ConsumerManager] 队列不存在！queueName=" + queueName);
        }
        //创建消费者对象
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
       synchronized (queue){
           queue.addConsumerEnv(consumerEnv);
           //消费对象 即从队列中取出message 如果队列中有消息 先消费队列中的消息
           int n = parent.getMemoryDataCenter().getMessageCount(queueName);

           //for循环结构消费消息
           for (int i = 0; i < n; i++) {
               //调用一次 消费一次
               consumeMessage(queue);
           }
       }


    }


    /**
     * 通知消费者可以进行消费了
     * @param queueName
     * @throws InterruptedException
     */
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }
   
}
